﻿using log4net.Appender;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using log4net.Core;
using KafkaNet;
using KafkaNet.Model;
using System.IO;
using log4net.Layout;
using HS.Log.Models;

namespace HS.Log
{
    internal class KafkaAppender : AppenderSkeleton
    {
        private readonly List<LogstashFileld> _fields = new List<LogstashFileld>();

        public string App { get; set; }

        private Producer _producer;

        public KafkaSettings KafkaSettings { get; set; }

        public override void ActivateOptions()
        {
            base.ActivateOptions();
            Start();
        }

        public void AddField(LogstashFileld fileld)
        {
            _fields.Add(fileld);
        }
        private void Start()
        {
            try
            {
                if (KafkaSettings == null) throw new LogException("KafkaSettings is missing");

                if (KafkaSettings.Brokers == null || KafkaSettings.Brokers.Count == 0) throw new Exception("Broker is not found");

                if (_producer == null)
                {
                    var brokers = KafkaSettings.Brokers.Select(x => new Uri(x)).ToArray();
                    var kafkaOptions = new KafkaOptions(brokers);
//#if DEBUG
//                    kafkaOptions.Log = new ConsoleLog();
//#else
				kafkaOptions.Log = new KafkaLog();
//#endif
                    _producer = new Producer(new BrokerRouter(kafkaOptions));
                    _producer.BatchSize = KafkaSettings.BatchSize;
                }
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("could not stop producer", ex);
            }

        }
        private void Stop()
        {
            try
            {
                _producer?.Stop();
            }
            catch (Exception ex)
            {
                ErrorHandler.Error("could not start producer", ex);
            }
        }
        private string GetTopic(LoggingEvent loggingEvent)
        {
            string topic = null;
            if (KafkaSettings.Topic != null)
            {
                var sb = new StringBuilder();
                using (var sw = new StringWriter(sb))
                {
                    KafkaSettings.Topic.Format(sw, loggingEvent);
                    topic = sw.ToString();
                }
            }

            if (string.IsNullOrEmpty(topic))
            {
                topic = $"{loggingEvent.LoggerName}.{loggingEvent.Level.Name}";
            }

            return topic;
        }
        private string GetMessage(LoggingEvent loggingEvent)
        {
            return loggingEvent.GetEvent(_fields, App);
        }

        protected override void Append(LoggingEvent loggingEvent)
        {
            var message = GetMessage(loggingEvent);
            var topic = GetTopic(loggingEvent);

            _producer.SendMessageAsync(topic, new[] { new KafkaNet.Protocol.Message(message) });
        }
        protected override void OnClose()
        {
            base.OnClose();
            Stop();
        }
    }

    public class KafkaSettings
    {
        public List<string> Brokers { get; set; }

        /// <summary>
        /// Get or set the message topic (routing key).  Default value is "%level". 
        /// </summary>
        public PatternLayout Topic { get; set; }

        /// <summary>
        /// BatchSize
        /// </summary>
        public int BatchSize { get; set; } = 100;
    }

    

}
